package com.lvmama.finance.demo.elasticsearch.sync;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import com.lvmama.finance.demo.elasticsearch.domain.jpa.NodeStatusEntity;
import com.lvmama.finance.demo.elasticsearch.domain.jpa.SyncRecordEntity;
import com.lvmama.finance.demo.elasticsearch.repository.jpa.NodeStatusRepository;
import com.lvmama.finance.demo.elasticsearch.repository.jpa.SyncRecordRepository;
import com.lvmama.finance.demo.elasticsearch.utils.InetAddressUtilities;

@Service
public class DataSyncManager {

	private static final Logger LOGGER = LoggerFactory.getLogger(DataSyncManager.class);

	@Autowired
	private SyncRecordRepository syncRecordRepository;
	@Autowired
	private NodeStatusRepository nodeStatusRepository;
	@Autowired
	@Qualifier("productSyncModule")
	private SyncModule productSyncModule;

	private static final Integer BATCH_SIZE = 500;
	private static final String APP_STATUS_ACTIVE = "active";
	private static final String APP_STATUS_STANDBY = "standby";
	private static final Integer APP_ACTIVE_PERIOD = 1000 * 60 * 3; // 3 minutes
	
	private static final Map<String, SyncModule> syncModuleMap = new HashMap<>();

	@PostConstruct
	public void init() {
		syncModuleMap.put("product", productSyncModule);
	}
	
	public void syncData() {
		if (!shouldSync()) {
			return ;
		}
		
		LOGGER.info("start to sync data");
		
		Page<SyncRecordEntity> page = null;

		try {
			for (int pageNo=0 ;; pageNo++) {
				page = syncRecordRepository.findAll(new PageRequest(pageNo, BATCH_SIZE));
				if (page.getContent().isEmpty()) {
					break;
				}
				syncData(page.getContent());
				
				if (!page.hasNext()) {
					break;
				}
				
				try {
					Thread.sleep(1000); // sleep 1s
				} catch (InterruptedException ignore) {
				}
			}
		} catch (Exception e) {
			LOGGER.error("sync data got failed", e);
		}
	}
	
	public void updateNodeStatus() {
		LOGGER.info("start to update node status");
		
		List<NodeStatusEntity> appStatusList = nodeStatusRepository.findByStatus(APP_STATUS_ACTIVE);
		
		NodeStatusEntity activeApp = null;
		
		for (NodeStatusEntity appStatus : appStatusList) {
			if (APP_STATUS_ACTIVE.equals(appStatus.getStatus()) && isUpdateToDate(appStatus.getLastUpdateTime())) {
				activeApp = appStatus;
				break;
			}
		}
		
		String localIp = InetAddressUtilities.getLocalIp();
		
		if (activeApp != null) { // found active application, just update time
			NodeStatusEntity status = null;
			if (activeApp.getNodeIp().equals(localIp)) {
				status = activeApp;
			} else {
				status = nodeStatusRepository.findOne(localIp);
				if (status == null) {
					status = new NodeStatusEntity();
					status.setNodeIp(localIp);
				}
				status.setStatus(APP_STATUS_STANDBY); 
			}
			status.setLastUpdateTime(new Date());
			nodeStatusRepository.save(status);
		} else {
			toBeActiveNode(localIp);
		}
	}
	
	@Transactional
	private void toBeActiveNode(String ip) {
		LOGGER.info("start elect current node to be active");
		
		boolean foundLocalNode = false;
		List<NodeStatusEntity> appStatusList = nodeStatusRepository.findAll();
		
		for (NodeStatusEntity entity : appStatusList) {
			if (!entity.getNodeIp().equals(ip)) {
				entity.setStatus(APP_STATUS_STANDBY);
			} else {
				entity.setStatus(APP_STATUS_ACTIVE);
				foundLocalNode = true;
			}
			entity.setLastUpdateTime(new Date());
		}
		nodeStatusRepository.save(appStatusList);
		
		if (!foundLocalNode) {
			NodeStatusEntity status = new NodeStatusEntity();
			status.setNodeIp(ip);
			status.setStatus(APP_STATUS_ACTIVE);
			status.setLastUpdateTime(new Date());
			nodeStatusRepository.save(status);
		}
		
	}
	
	private boolean isUpdateToDate(Date lastUpdateTime) {
		if (lastUpdateTime == null) 
			return false;
		return System.currentTimeMillis() - lastUpdateTime.getTime() < APP_ACTIVE_PERIOD;
	}
	
	protected boolean shouldSync() {
		try {
			String ip = InetAddressUtilities.getLocalIp();
			NodeStatusEntity status = nodeStatusRepository.findOne(ip);
			if (status != null && APP_STATUS_ACTIVE.equals(status.getStatus())) 
				return true;
		} catch (Exception e) {
			LOGGER.error("Exception when check app status", e);
		}
		return false;
	}

	private void syncData(List<SyncRecordEntity> syncRecords) {
		if (CollectionUtils.isEmpty(syncRecords))
			return;

		for (SyncRecordEntity syncRecord : syncRecords) {
			long total = syncData(syncRecord);
			LOGGER.info("sysc data of {}, last sync time is {}, total affected count is {}",
					syncRecord.getTableName(), syncRecord.getLastSyncTime(), total);
		}
	}

	private long syncData(SyncRecordEntity syncRecord) {
		SyncModule syncModule = syncModuleMap.get(syncRecord.getTableName());
		if (syncModule == null) 
			throw new RuntimeException("invalid table name found: " + syncRecord.getTableName());

		long total = 0;
		long start = System.currentTimeMillis();
		
		try {
			Date lastSyncTime = syncRecord.getLastSyncTime();
			if (lastSyncTime == null) 
				lastSyncTime = new Date(0);
			total += syncModule.doSync(lastSyncTime);
			
			syncRecord.setLastSyncTime(new Date());
			syncRecordRepository.save(syncRecord);
		} catch (Exception e) {
			LOGGER.error("do sync failed for {}", syncRecord.getTableName(), e);
		} finally {
			LOGGER.info("sync data got finished, takes {} seconds", (System.currentTimeMillis() - start) / 1000);
		}
		
		return total;
	}
}
